package com.example.filter;

import com.example.service.TPService;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER;

/**
 * @author zhouyp
 * @program tp-monitor
 * @description
 * @create 2020-07-09
 */
@Activate(group = CONSUMER)
public class TPMonitorFilter implements Filter {
	private volatile long start;
	private static final long oneMin = 60L;

	private final PriorityBlockingQueue<Integer> priorityQueueMethodA =
			new PriorityBlockingQueue<>(600,
					(t1, t2) -> t2 - t1
			);

	private final PriorityBlockingQueue<Integer> priorityQueueMethodB =
			new PriorityBlockingQueue<>(600,
					(t1, t2) -> t2 - t1
			);

	private final PriorityBlockingQueue<Integer> priorityQueueMethodC =
			new PriorityBlockingQueue<>(600,
					(t1, t2) -> t2 - t1
			);

	private final ScheduledExecutorService printExecutor = Executors
			.newScheduledThreadPool(1);

	private final ScheduledExecutorService clearExecutor = Executors
			.newScheduledThreadPool(1);

	private volatile boolean isSchedulerInit = false;

	private final AtomicInteger reqCount = new AtomicInteger(0);

	@Override
	public Result invoke(Invoker<?> invoker, Invocation invocation)
			throws RpcException {
		Result result;
		if (invoker.getInterface().equals(TPService.class)) {
			if (!isSchedulerInit) {
				initScheduler();
			}
			long start = System.currentTimeMillis();
			result = invoker.invoke(invocation);
			result.getValue();
			long end = System.currentTimeMillis();

			if (RpcContext.getContext().isAsyncStarted()) {
				return result;
			}
			reqCount.incrementAndGet();
			if (invocation.getMethodName().equals("methodA")) {
				priorityQueueMethodA.add((int) (end - start));
			}
			if (invocation.getMethodName().equals("methodB")) {
				priorityQueueMethodB.add((int) (end - start));
			}
			if (invocation.getMethodName().equals("methodC")) {
				priorityQueueMethodC.add((int) (end - start));
			}
			String endFlag = RpcContext.getContext().getAttachment("end");
			if (Boolean.parseBoolean(endFlag)) {
				System.out.println("--end consume from filter--");
				printExecutor.shutdown();
				clearExecutor.shutdown();
			}
		} else {
			result = invoker.invoke(invocation);
		}
		return result;
	}

	private void initScheduler() {
		this.start = System.currentTimeMillis();

		printExecutor.scheduleWithFixedDelay(() -> {
			printTPXX(priorityQueueMethodA, "methodA");
			printTPXX(priorityQueueMethodB, "methodB");
			printTPXX(priorityQueueMethodC, "methodC");
			System.out.println(" total request count is:" + reqCount.get());
		}, 5L, 5L, TimeUnit.SECONDS);

		clearExecutor.scheduleAtFixedRate(() -> {
			long now = System.currentTimeMillis();
			if (now - start > oneMin) {
				priorityQueueMethodA.clear();
				priorityQueueMethodB.clear();
				priorityQueueMethodC.clear();
				start = System.currentTimeMillis();
			}
			System.out.println("一分钟时间已过，清空消耗时间");
		}, 0L, oneMin, TimeUnit.SECONDS);

		isSchedulerInit = true;
	}

	private void printTPXX(PriorityBlockingQueue<Integer> queue, String methodName) {
		final List<Integer> tp90List = new ArrayList<>();
		final List<Integer> tp99List = new ArrayList<>();

		PriorityBlockingQueue<Integer> copiedQueue = new PriorityBlockingQueue<>(queue);

		for (int i = 0; i < getTP90Size(queue); i++) {
			int ele = copiedQueue.poll();
			if (i == getTP99Size(queue)) {
				tp99List.add(ele);
			}
			if (i == getTP90Size(queue) - 1) {
				tp90List.add(ele);
			}
		}

		final Date date = new Date();
		System.out.println(
				String.format("[%s]--[%s]-TP99 length:%s, TP99的耗时:%s", date, methodName, queue.size(), tp99List));
		System.out.println(
				String.format("[%s]--[%s]-TP90 length:%s, TP90的耗时:%s", date, methodName, queue.size(), tp90List));
	}

	private int getTP90Size(PriorityBlockingQueue<Integer> queue) {
		int i = (int) (queue.size() * 0.1);
		return i == 0 ? 1 : i;
	}

	private int getTP99Size(PriorityBlockingQueue<Integer> queue) {
		int i = (int) (queue.size() * 0.01);
		return i == 0 ? 1 : i;
	}

}
